package com.stone.im.dispatcher;

import com.alibaba.fastjson.*;
import com.google.protobuf.InvalidProtocolBufferException;
import com.stone.im.common.*;
import com.stone.im.protobuf.proto.AuthenticateRequestProto;
import com.stone.im.protobuf.proto.AuthenticateResponseProto;
import com.stone.im.protobuf.proto.MessagePushResponseProto;
import com.stone.im.protobuf.proto.MessageSendRequestProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.SocketChannel;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import redis.clients.jedis.Jedis;


/**
 * 分发系统处理类
 */
public class DispatcherHandler extends ChannelInboundHandlerAdapter {

    /**
     * 接收到一个接入系统发送过来的请求
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        SocketChannel channel = (SocketChannel)ctx.channel();

        Message message = new Message((ByteBuf) msg);
        System.out.println("收到一条消息，消息类型为：" + message.getMessageType());

        if(message.getMessageType() == Constants.MESSAGE_TYPE_REQUEST) {
            Request request = message.toRequest();

            if(request.getRequestType() == Constants.REQUEST_TYPE_AUTHENTICATE) {
                authenticate(ctx,request);
            }else if (message.getMessageType() == Constants.REQUEST_TYPE_SEND_MESSAGE){
                sendMessage(ctx,request);
            }
        }else if (message.getMessageType() == Constants.MESSAGE_TYPE_RESPONSE){
            Response response = message.toResponse();

            if (response.getRequestType() == Constants.REQUEST_TYPE_PUSH_MESSAGE){
                pushMessageResponse(response);
            }
        }
    }

    /**
     * 推送消息的响应
     * @param response
     */
    private void pushMessageResponse(Response response) throws InvalidProtocolBufferException {
        MessagePushResponseProto.MessagePushResponse messagePushResponse = MessagePushResponseProto.MessagePushResponse.parseFrom(response.getBody());

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("messageId",messagePushResponse.getMessageId());
        //TODO

        KafkaMenager instance = KafkaMenager.getInstance();
        KafkaProducer producer = instance.getProducer();

        ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                "push_message_response",
                jsonObject.toJSONString());

        producer.send(record);

    }

    private void authenticate(ChannelHandlerContext ctx,Request request) throws Exception {

        RequestHandler requestHandler = RequestHandler.getInstance();
        // 在这里应该是去找SSO单点登录系统去对用户的token进行认证
        AuthenticateRequestProto.AuthenticateRequest authenticateRequest =
                AuthenticateRequestProto.AuthenticateRequest.parseFrom(request.getBody());
        System.out.println("收到TCP接入系统发送的认证请求：" + authenticateRequest);

        AuthenticateResponseProto.AuthenticateResponse authenticateResponse =
                requestHandler.authenticate(authenticateRequest);

        if(authenticateResponse.getStatus() == Constants.RESPONSE_STATUS_OK) {
            SocketChannel socketChannel = (SocketChannel) ctx.channel();
            String gatewayChannelId = socketChannel.remoteAddress().getHostName() + ":"
                    + socketChannel.remoteAddress().getPort();

            String key = "session_"+authenticateRequest.getUid();
            // 其实在这里应该把session信息写入Redis的
            String value="{"
                    +"token:"+authenticateRequest.getToken()+","
                    +"timestamp:"+authenticateRequest.getTimestamp()+","
                    +"isAuthenticated：true,"+
                    "authenticateTimestamp:"+System.currentTimeMillis()+","+
                    "gatewayChannelId:"+gatewayChannelId+ "}";

            JedisManager jedisManager = JedisManager.getInstance();
            Jedis jedis = jedisManager.getJedis();
            jedis.set(key,value);

            System.out.println("在Redis中写入分布式Session......");
        }

        Response response = new Response(request, authenticateResponse.toByteArray());
        ctx.writeAndFlush(response.getBuffer());

        System.out.println("返回响应给TCP接入系统：" + authenticateResponse);
    }

    private void sendMessage(ChannelHandlerContext ctx,Message message) throws Exception {
        SocketChannel channel = (SocketChannel)ctx.channel();
        String channelId = channel.remoteAddress().getHostName()+":"+channel.remoteAddress().getPort();
        Request request = message.toRequest();
        KafkaMenager instance = KafkaMenager.getInstance();
        KafkaProducer producer = instance.getProducer();
        MessageSendRequestProto.MessageSendRequest messageSendRequest =
                MessageSendRequestProto.MessageSendRequest.parseFrom(request.getBody());

        JSONObject jsonObject = new JSONObject();
        jsonObject.put("senderId",messageSendRequest.getSenderId());
        jsonObject.put("receiverId",messageSendRequest.getReceiverId());
        jsonObject.put("content",messageSendRequest.getContent());
        jsonObject.put("requestType",Constants.REQUEST_TYPE_SEND_MESSAGE);
        jsonObject.put("sequence",message.getSequence());
        jsonObject.put("gatewayChannelId",channelId);
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                "send_message",
                jsonObject.toJSONString());

        producer.send(record);
    }

    /**
     * 建立连接
      * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        SocketChannel channel = (SocketChannel) ctx.channel();
        String channelId = channel.remoteAddress().getHostName() +":"+channel.remoteAddress().getPort();
        GatewayInstanceManager instance = GatewayInstanceManager.getInstance();
        instance.addGatewayInstance(channelId,channel);
        System.out.println("已经跟tcp接入系统建立连接，地址为:"+channel.remoteAddress().getAddress());
    }

    /**
     * 断开连接
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        SocketChannel channel = (SocketChannel) ctx.channel();
        GatewayInstanceManager instance = GatewayInstanceManager.getInstance();
        String channelId = channel.remoteAddress().getHostName() +":"+channel.remoteAddress().getPort();
        instance.removeGatewayInstance(channelId);
        System.out.println("跟tcp系统连接断开");
    }

    /**
     * 处理完请求
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 发生异常
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}
